package com.ge.mythreed.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @Author geqingxiang
 * @PackageName my-project-test
 * @Package com.ge.threed.consumer
 * @Date 2023/12/12 17:52
 * @Version 1.0
 */
@Slf4j
@Component
public class KafkaConsumer {
    /**
     * 监听消费
     * @param record
     */
    @KafkaListener(topics = {"sb_topic"})
    public void onNormalMessage(ConsumerRecord<String, Object> record) {
        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" +
                record.value());
    }

//    @KafkaListener(topics = "canalKafkaTopicUserCenter", groupId = "canalKafkaTopic")
    @KafkaListener(topics = "canalKafkaTopicUserCenter")
    public void canalKafkaTopic001(ConsumerRecord<String, String> record) {
        //数据库变动的值
        String value = record.value();
        log.info("数据表发生变动:{}", value);
    }

    @KafkaListener(topics = {"kafka_consumer_binlog"})
    public void onNormalMessageTwo(ConsumerRecord<String, Object> record) {
        System.out.println("测试消费：" + record.topic() + "-" + record.partition() + "=" +
                record.value());
    }
}
